import com.cw.recommend.common.Runner.{RunnerConfig, runSpark}
import com.cw.recommend.common.constant._
import com.cw.recommend.common.model.ProductSimList
import com.cw.recommend.common.util.{MongoDBUtil, RedisUtil}
import com.mongodb.casbah.commons.MongoDBObject
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.streaming.kafka010.{ConsumerStrategies, KafkaUtils, LocationStrategies}
import org.apache.spark.streaming.{Seconds, StreamingContext}

object RealtimeRecommend {


  def main(args: Array[String]): Unit = {

    implicit val conf = RunnerConfig("realtime_recommend")

    runSpark { spark =>
      val sc = spark.sparkContext
      val ssc = new StreamingContext(sc, Seconds(2))

      import spark.implicits._

      // 读取物品相似度矩阵
      val productSimMTX = MongoDBUtil
        .readMongoDB(spark, PRODUCT_SIM_LIST)
        .as[ProductSimList]
        .rdd
        .map(simList =>
          simList.productId ->
            simList.itemList.map(item => (item.productId, item.sim)).toMap
        )
        .collectAsMap()

      val productSimListBC = sc.broadcast(productSimMTX)


      // 读取kafka中的实时评分数据
      val kafkaParams = Map(
        "bootstrap.servers" -> KAFKA_BROKERS,
        "key.deserializer" -> classOf[StringDeserializer],
        "value.deserializer" -> classOf[StringDeserializer],
        "group.id" -> "recommender",
        "auto.offset.reset" -> "latest"
      )

      val kafkaStream = KafkaUtils.createDirectStream(
        ssc,
        LocationStrategies.PreferConsistent,
        ConsumerStrategies.Subscribe[String, String](Seq(RECOMMENDER_TOPIC), kafkaParams)
      )

      val ratingStream = kafkaStream.map { record =>
        // record format:   userID|productId|score|timestamp
        val splits = record.value.split("\\|")
        println(s"当前评分: ${splits.toSeq}")
        (splits(0).trim.toInt, splits(1).trim.toInt, splits(2).trim.toDouble, splits(3).trim.toLong)
      }

      ratingStream.foreachRDD { rdd =>
        rdd.foreach { case (userId, productId, score, timestamp) =>
          // redis中最近评分
          val recentRantings = queryRecentRatings(userId)

          // 从物品相似矩阵中获得候选物品
          val candidates = getCandidateProducts(productId, productSimListBC)
          println(s"候选物品: ${candidates.toIndexedSeq}")

          // 计算候优先级高的物品
          val recommendList = highPriorProducts(candidates, recentRantings, productSimListBC)

          // 写入mongodb
          println(s"推荐列表${recommendList.toIndexedSeq}")
          saveMongodb(userId, recommendList)

        }
      }

      ssc.start()
      ssc.awaitTermination()

    }


  }

  def queryRecentRatings(userId: Int): Array[(Int, Double)] = {
    import collection.JavaConversions._
    RedisUtil.jedisContext { j =>
      // key> userId:xx     productId:score
      j.lrange(s"userId:$userId", 0, RECENTLY_NUM - 1)
        .map { s =>
          val splits = s.split(":")
          splits(0).trim.toInt -> splits(1).trim.toDouble
        }
        .toArray
    }
  }


  private type SimilarItems = Map[Int, Double]

  def getCandidateProducts(userId: Int, simListBC: Broadcast[collection.Map[Int, SimilarItems]]): Array[Int] = {
    simListBC.value.get(userId) match {
      case Some(items) =>
        val candidates = items.toArray
        // 拉取用户已经评分过的物品
        val historyRatings = MongoDBUtil.clientContext { cli =>
          cli(MONGODB_DB)(RATING_COLLECTION)
            .find(MongoDBObject("userId" -> userId))
            .map(_.get("productId").toString.toInt)
            .toArray
        }
        candidates.filter(tp => !historyRatings.contains(tp._1))
          .sortWith(_._2 > _._2)
          .take(MAX_CANDIDATE_NUM)
          .map(_._1)
      case None => Array.empty[Int]
    }

  }

  def highPriorProducts(candidates: Array[Int], recentRantings: Array[(Int, Double)], simListBC: Broadcast[collection.Map[Int, SimilarItems]]) = {
    val simList = simListBC.value
    candidates.flatMap { productId =>
        val similarRatings = recentRantings.flatMap { case (recentProductId, score) =>
          val similarity = sim(recentProductId, productId, simList)
          if (similarity >= 0.6) Some((recentProductId, score, similarity)) else None
        }
        if (similarRatings.length > 0) Some(productId -> similarRatings) else None
      }.map { case (productId, similarRatings) =>
        val sum = similarRatings.foldLeft(0D)((acc, tp3) => acc + tp3._2 * tp3._3) // 相似权重和分数相乘，再求和
        val num = similarRatings.length
        val enhance = similarRatings.count(_._2 >= 3) // 与当前候选相似的最近评分中分数高于3的
        productId -> (sum / num + log(enhance.max(1)) - log((num - enhance).max(1)))
      }.sortBy(_._2)(Ordering[Double].reverse)
      .take(RECOMMEND_NUM)
  }



  def saveMongodb(userId: Int, recommendList: Array[(Int, Double)]) = {
    MongoDBUtil.clientContext {cli =>
      val coll = cli(MONGODB_DB)(REALTIME_RECOMMEND_LIST)
      coll.findAndRemove(MongoDBObject("userId" -> userId))
      coll.insert(MongoDBObject(
        "userId"->userId,
        "recs" -> recommendList.map(x => MongoDBObject(
          "productId"->x._1, "score"->x._2
        ))
      ))
    }
  }






  private def log(x: Int, base: Int = 10) = math.log(x) / math.log(10)


  private def sim(productId1: Int, productId2: Int, simList: collection.Map[Int, SimilarItems]) = {
    simList.get(productId1) match {
      case Some(x) => x.get(productId2) match {
        case Some(sim) => sim
        case None => 0D
      }
      case None => 0D
    }

  }


}
